{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "netcat: Swiss army knife pf networking"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Why netcat?"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- Read/write two (three) file descriptors simultaneously\n",
    "- Two simple concurrent modes are examined\n",
    "  - Thread-per-connectionwith blocking IO\n",
    "  - IO-multiplexing with non-blocking IO\n",
    "- How/when to close a TCP connection?\n",
    "  - Ensure all data aresent and received\n",
    "- Why IO-multiplexing must be used with non-blocking IO?"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 读写socket文件描述符和标准输入输出文件stdin/stdout描述符\n",
    "- 两种基本的并发模型: 一个线程一个连接; IO复用\n",
    "- 如何安全的关闭TCP连接; TTCP协议设计上避免连接关闭的问题; netcat重新检视这个问题\n",
    "- IO复用为什么与非阻塞IO配合"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Swiss army knife"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- `nc < /dev/zero` == chargen\n",
    "- `nc > /dev/null` == discard\n",
    "- `dd /dev/zero | nc` == poor man's ttcp\n",
    "- `nc < file, nc > file` == poor man's scp\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 信号发生器, 发生大量数据\n",
    "- 负载, 接收大量数据\n",
    "- 产生固定量的数据\n",
    "- 两台机器之间拷贝文件; 开销小于scp, 没有加密解密; 如何保证收到的文件与原文件一致, 计算md5, TCP可靠一般不会有错, 但是socket编程可能存在问题"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## How/when to close a TCP connection?"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 建立TCP的难度小于销毁TCP的难度; 服务端建立TCP的难度小于客户端建立连接的难度; 接受TCP的难度小于发送TCP的难度"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "[The ultimate SO_LINGER page, or: why is my tcp not reliable](https://blog.netherlabs.nl/articles/2009/01/18/the-ultimate-so_linger-page-or-why-is-my-tcp-not-reliable)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- Why my TCP is not reliable?\n",
    "  - Sent all then close, but lastfew bytes are lost.\n",
    "\n",
    "- Wrong: send() + close()\n",
    "  - When there is data in input buffer, close() will cause RST, terminate connection prematurelyWith or without SO_LINGER? TL;DR: don't use LINGER"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 如何正确关闭TCP连接, 保证对方已经收到全部数据, 初学者常犯错误:写一个传输文件程序, send之后直接close, 之后发现有时候收到的文件比原文件小, 最后几十k数据丢失了\n",
    "- 链接页面: 解释了, 协议栈接受缓冲区中有数据, 程序未读, 直接调用close, 会导致TCP协议栈发送reset分解, 强行断开连接, 如果协议栈发送缓冲区还有数据对方没有收到这个数据, 数据就丢失了; `SO_LINGER`并不能解决这个问题"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "// https://github.com/chenshuo/recipes/blob/master/tpc/bin/sender.cc\n",
    "\n",
    "\n",
    "void sender(const char* filename, TcpStreamPtr stream)\n",
    "{\n",
    "  FILE* fp = fopen(filename, \"rb\");\n",
    "  if (!fp)\n",
    "    return;\n",
    "\n",
    "  printf(\"Sleeping 10 seconds.\\n\");\n",
    "  sleep(10);\n",
    "\n",
    "  printf(\"Start sending file %s\\n\", filename);\n",
    "  char buf[8192];\n",
    "  size_t nr = 0;\n",
    "  while ( (nr = fread(buf, 1, sizeof buf, fp)) > 0)\n",
    "  {\n",
    "    stream->sendAll(buf, nr);\n",
    "  }\n",
    "  fclose(fp);\n",
    "  printf(\"Finish sending file %s\\n\", filename);\n",
    "\n",
    "  // 正确关闭TCP连接的逻辑, 注释重现TCP\"不可靠\"的现象\n",
    "  // Safe close connection\n",
    "  printf(\"Shutdown write and read until EOF\\n\");\n",
    "  stream->shutdownWrite(); // 关闭TCP连接的写端\n",
    "  while ( (nr = stream->receiveSome(buf, sizeof buf)) > 0) // 一直读数据直到读取0(end of file)读空缓冲区, 表明对方已关闭链接\n",
    "  {\n",
    "    // do nothing\n",
    "  }\n",
    "  printf(\"All done.\\n\");\n",
    "\n",
    "  // TcpStream destructs here, close the TCP socket.\n",
    "  // 对象析构自动调用close\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "```shell\n",
    "./sender ttcp 1234\n",
    "```\n",
    "\n",
    "```shell\n",
    "nc localhost 1234 | wc -c\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "当`nc`带有数据发送给sender, sender没有去读, 造成sender进行close时没有正常的FIN分解, 而是RST, 直接强行终止连接"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "对于接受方, read返回0, 就是遇到了end of file, 然后可以close; 总结, read返回0, 才会close.\n",
    "\n",
    "发送方shut down write 会导致对方(接收方)的read返回0, 对方会关闭连接; 自己的socket的read也会返回零, 于是自己也关闭连接"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> 之后有时间进行实验"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- Correct sender: send() + shutdown(WR) + read() ->0 + close()\n",
    "\n",
    "- Correct receiver: read() -> 0 + if nothing more to send + close()\n",
    "\n",
    "- Or, better, design your app protocol to allow safedisconnecting"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 对于发送方, send完毕之后, 调用shut down write; 对于接收方, read返回0, 然后接收方没有其他数据发送, 会close;对于发送方, read返回0, 然后close\n",
    "- 一般read返回0 ,才会close, 如果遇到恶意或者有BUG的客户端, 可能会一直不close, 因为会加个超时控制, shut down write之后若干时间后, 还没有发生read返回0的事件, 强行直接关闭连接, 这是为了security, 不是数据的safety或者完整性的integrity; 我们一般依赖TCP shut down write会发送FIN(告诉对方end of file), 数据已经发送完了, 更好的办法是设计协议的时候, 包含了数据的长度, 这样接收方可以主动判断数据是否已经收全了, 不必依赖end of file, 例如之前的TTCP例子\n",
    "- 就算采用了推荐的方法, 发送方也无法确切知道接收方已经接收到了数据, 但是这个是在不修改协议的情况下, 这是发送发尽到的最大努力"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Ignore `SIGPIPE` in any concurrent TCP server"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- SIGPIPE is sent to writer when the other end of \"pipe\" is closed\n",
    "  - Default action is terminating the process, good for command line piping\n",
    "    - gunzip -c huge.log.gz | grep ERROR | head"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 如果程序向一个对方已经关闭的管道写数据, 会收到`SIGPIPE`信号, 收到`SIGPIPE`信号的默认行为是终止进程; \n",
    "- 在UNIX中的, 使用管道串联一些列命令, 管道末端的程序退出了, 管道中命令会依次收到`SIGPIPE`, 程序从而会依次退出\n",
    "  - `gunzip -c huge.log.gz | grep ERROR | head` : 解压缩一个大的日志文件, 找出包含ERROR的, 取前10行(由系统决定), 避免解压缩整个大文件\n",
    "  - UNIX默认阻塞IO, 这个起到了自动限速(节流)的作用; 如果接收端收的慢, 发送端阻塞在write上, CPU占用率下降; `gunzip`由于后面跟着管道, 限制了`gunzip`的解压缩速度, 降低了CPU占用"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- What if clientcloses a socket when server is still writing to it?\n",
    "  - Client could be offensive or misbehaving, server should be defensive\n",
    "- The default action willkill the server and affect all clients\n",
    "  - Where is my server process after Ctrl-C a client?"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 到了网络时代, `SIGPIPE`行为就有副作用了, 如果对方关闭了socket, 你还要往上写数据, 同样会收到`SIGPIPE`信号; 如果不做特殊处理, 任何一个客户端意外关闭连接, 服务进程就会收到`SIGPIPE`而退出, 会影响其他的客户使用; 因此网络程序一般会在进程启动时忽略`SIGPIPE`信号;"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- So ignore SIGPIPE in network programs\n",
    "  - If a program writes to stdout, check return value of printf() then exit()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 如果网路程序也是使用管道\"串\"起来, 例如实现了一个curl, 如果忽略了`SIGPIPE`, 使用`printf`之类的函数往标准输出打印东西时, 要检查它的返回值, 如果对方关闭, `printf`会返回一个错误码而不会收到`SIGPIPE`, 让你退出, 检查返回值, 如果小于`0`, 表明管道的另外一端已经关闭, 我们应该主动终止程序"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Nagle's algorithm, TCP_NODELAY"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- Affect latency of request-response protocol\n",
    "- write() will not send data if there isany unacked TCP segment\n",
    "  - See [Nagle's_algorithm](https://en.wikipedia.org/wiki/Nagle's_algorithm) for full spec"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- TCP网络编程难度发送大于接收;\n",
    "- Nagle's algorithm/`TCP_NODELAY`: 如果已经发了数据, 在未收到响应之前, 再调用write发送数据, 会等待TCP响应/ACK返回, 再发送. 是为了, 程序写的有问题发送很多\"小\"的包, 造成网络负担. 但是这严重影响了请求响应式协议的延迟"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- For write-write-read,the second write will be delayed by one RTT\n",
    "  - Solution: buffering, make it write-read\n",
    "  - However, it still affect request pipelining"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 如果程序是\"write-write-read\": 第二个`write`会被退后一个往返延迟, 因为第一个`write`还没收到对方TCP的ACK\n",
    "  - 应用层做缓冲: 合并2个`write`变成`write-read`; 但\n",
    "  - request pipelining: 是当同一个连接上有多个请求, 不是发一个请求接收一个响应, 再发第二个请求, 而是发多个请求, 并发请求, 这种情况很难事前将并发请求合并成1个buff, 因为可能在程序的不同地方, 发送的请求. 这种情况下`TCP_NODELAY`会影响TCP响应的延迟, 原来可以发4个请求, 服务器发4个响应, 中间是1个round trip的延迟, 如果Nagle's algorithm生效, 则会变成多个, 造成多个更多的延迟"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- I recommend to disable Nagle’s algorithm by default\n",
    "  - Go does this for every TCP connection"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 建议关闭`TCP_NODELAY`选项"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 实验"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "为了观察这个延迟增加现象, 将服务端程序放到远程机器上(与运行机器往返延迟大约150ms)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "// 客户端 https://github.com/chenshuo/recipes/blob/master/tpc/bin/nodelay.cc\n",
    "\n",
    "#include \"InetAddress.h\"\n",
    "#include \"TcpStream.h\"\n",
    "\n",
    "#include <string.h>\n",
    "#include <sys/time.h>\n",
    "#include <unistd.h>\n",
    "\n",
    "double now()\n",
    "{\n",
    "  struct timeval tv = { 0, 0 };\n",
    "  gettimeofday(&tv, NULL);\n",
    "  return tv.tv_sec + tv.tv_usec / 1000000.0;\n",
    "}\n",
    "\n",
    "int main(int argc, char* argv[])\n",
    "{\n",
    "  if (argc < 3)\n",
    "  {\n",
    "    printf(\"Usage: %s [-b] [-D] [-n num] hostname message_length\\n\", argv[0]);\n",
    "    printf(\"  -b Buffering request before sending.\\n\"\n",
    "           \"  -D Set TCP_NODELAY.\\n\"\n",
    "           \"  -n num Send num concurrent requests, default = 1.\\n\");\n",
    "    return 0;\n",
    "  }\n",
    "\n",
    "  int opt = 0;\n",
    "  bool buffering = false;\n",
    "  bool tcpnodelay = false;\n",
    "  int num = 1;\n",
    "  while ( (opt = getopt(argc, argv, \"bDn:\")) != -1)\n",
    "  {\n",
    "    switch (opt)\n",
    "    {\n",
    "      case 'b':\n",
    "        buffering = true; // 只发一次数据\n",
    "        break;\n",
    "      case 'D':\n",
    "        tcpnodelay = true; // 开启TCP_NODELAY\n",
    "        break;\n",
    "      case 'n':\n",
    "        num = atoi(optarg);\n",
    "        break;\n",
    "      default:\n",
    "        printf(\"Unknown option '%c'\\n\", opt);\n",
    "        return 0;\n",
    "    }\n",
    "  }\n",
    "\n",
    "  if (optind > argc - 2)\n",
    "  {\n",
    "    printf(\"Please specify hostname and message_length.\\n\");\n",
    "    return 0;\n",
    "  }\n",
    "\n",
    "  const char* hostname = argv[optind];\n",
    "  int len = atoi(argv[optind+1]);\n",
    "\n",
    "  const uint16_t port = 3210;\n",
    "  InetAddress addr;\n",
    "  if (!InetAddress::resolve(hostname, port, &addr))\n",
    "  {\n",
    "    printf(\"Unable to resolve %s\\n\", argv[1]);\n",
    "    return 0;\n",
    "  }\n",
    "\n",
    "  printf(\"connecting to %s\\n\", addr.toIpPort().c_str());\n",
    "  TcpStreamPtr stream(TcpStream::connect(addr));\n",
    "  if (!stream)\n",
    "  {\n",
    "    printf(\"Unable to connect %s\\n\", addr.toIpPort().c_str());\n",
    "    perror(\"\");\n",
    "    return 0;\n",
    "  }\n",
    "\n",
    "  if (tcpnodelay)\n",
    "  {\n",
    "    stream->setTcpNoDelay(true); // 开启TCP_NODELAY\n",
    "    printf(\"connected, set TCP_NODELAY\\n\");\n",
    "  }\n",
    "  else\n",
    "  {\n",
    "    stream->setTcpNoDelay(false);\n",
    "    printf(\"connected\\n\");\n",
    "  }\n",
    "\n",
    "\n",
    "  double start = now();\n",
    "  for (int n = 0; n < num; ++n)\n",
    "  {\n",
    "    printf(\"Request no. %d, sending %d bytes\\n\", n, len);\n",
    "    if (buffering) // 准备好一个消息\n",
    "    {\n",
    "      std::vector<char> message(len + sizeof len, 'S'); // 命令行指定长度+4\n",
    "      memcpy(message.data(), &len, sizeof len);\n",
    "      int nw = stream->sendAll(message.data(), message.size());\n",
    "      printf(\"%.6f sent %d bytes\\n\", now(), nw);\n",
    "    }\n",
    "    else // 发两次\n",
    "    {\n",
    "      stream->sendAll(&len, sizeof len);\n",
    "      printf(\"%.6f sent header\\n\", now());\n",
    "      usleep(1000); // prevent kernel merging TCP segments; 阻止kernel将数据合并成一个segment\n",
    "      std::string payload(len, 'S');\n",
    "      int nw = stream->sendAll(payload.data(), payload.size());\n",
    "      printf(\"%.6f sent %d bytes\\n\", now(), nw);\n",
    "    }\n",
    "  }\n",
    "\n",
    "  printf(\"Sent all %d requests, receiving responses.\\n\", num);\n",
    "  for (int n = 0; n < num; ++n)\n",
    "  {\n",
    "    int ack = 0;\n",
    "    int nr = stream->receiveAll(&ack, sizeof ack);\n",
    "    printf(\"%.6f received %d bytes, ack = %d\\n\", now(), nr, ack);\n",
    "  }\n",
    "  printf(\"total %f seconds\\n\", now() - start); // 时间差\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "// 服务端 https://github.com/chenshuo/recipes/blob/master/tpc/bin/nodelay_server.cc\n",
    "\n",
    "#include \"Acceptor.h\"\n",
    "#include \"InetAddress.h\"\n",
    "#include \"TcpStream.h\"\n",
    "\n",
    "#include <thread>\n",
    "#include <vector>\n",
    "\n",
    "#include <assert.h>\n",
    "#include <string.h>\n",
    "#include <sys/time.h>\n",
    "\n",
    "double now()\n",
    "{\n",
    "  struct timeval tv = { 0, 0 };\n",
    "  gettimeofday(&tv, NULL);\n",
    "  return tv.tv_sec + tv.tv_usec / 1000000.0;\n",
    "}\n",
    "\n",
    "// an interative request-response server\n",
    "int main(int argc, char* argv[])\n",
    "{\n",
    "  InetAddress listenAddr(3210);\n",
    "  Acceptor acceptor(listenAddr);\n",
    "  printf(\"Accepting... Ctrl-C to exit\\n\");\n",
    "  int count = 0;\n",
    "  bool nodelay = argc > 1 && strcmp(argv[1], \"-D\") == 0;\n",
    "  while (true)\n",
    "  {\n",
    "    TcpStreamPtr tcpStream = acceptor.accept();\n",
    "    printf(\"accepted no. %d client\\n\", ++count);\n",
    "    if (nodelay)\n",
    "      tcpStream->setTcpNoDelay(true);\n",
    "\n",
    "    while (true)\n",
    "    {\n",
    "      int len = 0;\n",
    "      int nr = tcpStream->receiveAll(&len, sizeof len);\n",
    "      if (nr <= 0)\n",
    "        break;\n",
    "      printf(\"%f received header %d bytes, len = %d\\n\", now(), nr, len);\n",
    "      assert(nr == sizeof len);\n",
    "\n",
    "      std::vector<char> payload(len);\n",
    "      nr = tcpStream->receiveAll(payload.data(), len);\n",
    "      printf(\"%f received payload %d bytes\\n\", now(), nr);\n",
    "      assert(nr == len);\n",
    "      int nw = tcpStream->sendAll(&len, sizeof len);\n",
    "      assert(nw == sizeof len);\n",
    "    }\n",
    "\n",
    "    printf(\"no. %d client ended.\\n\", count);\n",
    "  }\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 客户端\n",
    "\n",
    "```shell\n",
    "./nodelay IP 1000\n",
    "```\n",
    "\n",
    "- 服务端\n",
    "\n",
    "```shell\n",
    "sudo tcpdump -i eth3 -n tcp port 3210\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 客户端: 开启`TCP_NODELAY`\n",
    "\n",
    "```shell\n",
    "./nodelay -D IP 1000\n",
    "```\n",
    "\n",
    "- 服务端\n",
    "\n",
    "```shell\n",
    "sudo tcpdump -i eth3 -n tcp port 3210\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 客户端: 使用`buffing`\n",
    "\n",
    "```shell\n",
    "./nodelay -b IP 1000\n",
    "```\n",
    "\n",
    "- 服务端\n",
    "\n",
    "```shell\n",
    "sudo tcpdump -i eth3 -n tcp port 3210\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "![4.1](./images/4.1.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> 一次完整的TCP三次握手过程，对应的是TCP连接的建立流程。\n",
    "> - 第一次握手：客户端 → 服务器，SYN\n",
    "> - 第二次握手：服务器 → 客户端，SYN+ACK\n",
    "> - 第三次握手：客户端 → 服务器，ACK"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> `11:55:48.047696 IP 104.155.216.238.3210 > 10.0.0.37.21153: Flags [P.], seq 1:5, ack 1005, win 236, options [nop,nop,TS val 13826262 ecr 1501596], length 4`\n",
    "> - `Flags [P.]`：带有数据推送（PSH）和确认（ACK）标志\n",
    "> - `seq 1:5, ack 1005`：序列号1到4，acknowledge对方1004\n",
    "> - `length 4`：负载4字节"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Trilogy of TCP client/server"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- SO_REUSEADDR\n",
    "  - So that a TCP server can restart immediately after crash/kill\n",
    "  - Also needed for fork-per-connection model"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "TCP客户端/服务端程序三部曲(规定动作)\n",
    "- 开启`SO_REUSEADDR`: 复用地址, TCP Server Crash后可以马上重启, 否则可能一段时间内, 不能在同一端口监听, 是为了防止有些packages还在网上传输\n",
    "- 如果使用fork-per-connection模型, fork之后, 别的进程也用了这个端口, 例如监听80端口, 连接进来后, fork到别的进程也用了80端口, 那么服务器lisening那个进程退出后, 你不能再起一个listener在80端口监听进程, 它会觉得你这个地址已经被用掉了"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- Ignore SIGPIPE"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- TCP_NODELAY"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Operation modes"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- Server\n",
    "  - bind + listen + accept\n",
    "\n",
    "- Client\n",
    "  - resolve address + connect\n",
    "\n",
    "- Once connection has been established, c/s behavethe same\n",
    "- Two parallel loops serve two directions:\n",
    "  - Read from stdin, write to TCP socket\n",
    "  - Read from TCP socket, write to stdout"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 实现netcat的基本功能\n",
    "- 服务端/客户端, 区别在于连接的建立方式, 一旦TCP连接建立起来, 程序的行为一模一样\n",
    "- 两个并行的循环: 两个方向的数据流动\n",
    "  - 标准输入(stdin)读, 写入socket; \n",
    "  - 从socket读, 写入标准输出(stdout)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "![4.2](./images/4.2.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 实验"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 服务端\n",
    "```shell\n",
    "nc -l 1234\n",
    "```\n",
    "\n",
    "- 客户端\n",
    "```shell\n",
    "nc localhost 1234\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> 输入发送内容后, 敲回车; `Ctrl + D`退出程序"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "vscode": {
     "languageId": "shellscript"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "12345\n"
     ]
    }
   ],
   "source": [
    "! nc -l 1234"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Not all netcats are created equal"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "| chargen in C++     | chargen in Python | chargen in muduo |\n",
    "| ------------------ | ----------------- | ---------------- |\n",
    "| netcat.cc          |                   |                  |\n",
    "| netcat.py          |                   |                  |\n",
    "| netcat-nonblock.py |                   |                  |\n",
    "| stock nc           |                   |                  |"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "| 路径                                   | 类型                |\n",
    "|:----------------------------------------|:--------------------|\n",
    "| [recipes/tpc/netcat.cc](https://github.com/chenshuo/recipes/blob/master/tpc/bin/netcat.cc)                   | thread-per-connection|\n",
    "| [recipes/python/netcat.py](https://github.com/chenshuo/recipes/blob/master/python/netcat.py)                | IO-multiplexing     |\n",
    "| [recipes/python/netcat-nonblock.py](https://github.com/chenshuo/recipes/blob/master/python/netcat-nonblock.py)       | IO-multiplexing     |"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- Load generator: chargen\n",
    "  - [recipes/tpc/chargen.cc](https://github.com/chenshuo/recipes/blob/master/tpc/bin/chargen.cc)\n",
    "  - [recipes/python/chargen.py](https://github.com/chenshuo/recipes/blob/master/python/chargen.py)\n",
    "  - [muduo/examples/simple/chargen/*](https://github.com/chenshuo/muduo/tree/master/examples/simple/chargen)\n",
    "\n",
    "> 负载生成器"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 实验"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 系统的`nc`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "```shell\n",
    "utop # 查看进程CPU使用率\n",
    "```\n",
    "\n",
    "```shell\n",
    "./chargen -l 1234 # 可以打印出吞吐量\n",
    "```\n",
    "\n",
    "```shell\n",
    "nc localhost 1234 > /dev/null\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### [recipes/tpc/netcat.cc](https://github.com/chenshuo/recipes/blob/master/tpc/bin/netcat.cc) "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "```shell\n",
    "utop # 查看进程CPU使用率\n",
    "```\n",
    "\n",
    "```shell\n",
    "./chargen -l 1234 # 可以打印出吞吐量\n",
    "```\n",
    "\n",
    "```shell\n",
    "netcat localhost 1234 > /dev/null\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> CPU率使用率高的进程, 拖慢了性能"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### [recipes/python/netcat.py](https://github.com/chenshuo/recipes/blob/master/python/netcat.py)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "```shell\n",
    "utop # 查看进程CPU使用率\n",
    "```\n",
    "\n",
    "```shell\n",
    "./chargen -l 1234 # 可以打印出吞吐量\n",
    "```\n",
    "\n",
    "```shell\n",
    "python netcat.py localhost 1234 > /dev/null\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### [recipes/python/netcat-nonblock.py](https://github.com/chenshuo/recipes/blob/master/python/netcat-nonblock.py) "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "```shell\n",
    "utop # 查看进程CPU使用率\n",
    "```\n",
    "\n",
    "```shell\n",
    "./chargen -l 1234 # 可以打印出吞吐量\n",
    "```\n",
    "\n",
    "```shell\n",
    "python netcat-nonblock.py localhost 1234 > /dev/null\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 使用`pv`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "```shell\n",
    "utop # 查看进程CPU使用率\n",
    "```\n",
    "\n",
    "```shell\n",
    "./chargen -l 1234 # 可以打印出吞吐量\n",
    "```\n",
    "\n",
    "```shell\n",
    "netcat localhost 1234 | pv > /dev/null # 吞吐量会下降\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## thread-per-connection VS IO-multiplexing"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "当并发连接数少时, thread-per-connection的性能优于IO-multiplexing的性能, thread-per-connection节约了了等待IO就绪事件的系统调用"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> - **thread-per-connection**：每来一个连接，创建一个独立线程处理。线程阻塞在IO操作上，等待数据。\n",
    "> - **IO-multiplexing**（如 `select`/`poll`/`epoll`）：单线程或少量线程，使用系统调用统一监听多个连接，哪个连接有IO事件就处理哪个。\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Thread-per-connection with blocking IO"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- Those two parallel loops map to two thread natively\n",
    "  - Thread-per-half-connection, to be exact\n",
    "  - Go uses this model universally"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 采用阻塞IO + thread-per-connection的并发模型; 每个连接对应2个线程, 每个线程只管半个连接, 要么收/读, 要么发/写, 这种模型在go语言中使用的最多\n",
    "  - 主线程负责读stdin, 写socket; 另外线程(网络线程)读socket, 写stdout\n",
    "\n",
    "> go语言: go runtime, channel, select; select其他语言很难模仿, 直接对应"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- Two directions are not affecting each other\n",
    "  - How to tell the other thread to exit, when it's blocking on reador write?\n",
    "- Blocking IO throttles traffic by default"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "// https://github.com/chenshuo/recipes/blob/master/tpc/bin/netcat.cc\n",
    "\n",
    "#include \"Acceptor.h\"\n",
    "#include \"InetAddress.h\"\n",
    "#include \"TcpStream.h\"\n",
    "\n",
    "#include <thread>\n",
    "\n",
    "#include <string.h>\n",
    "#include <unistd.h>\n",
    "\n",
    "int write_n(int fd, const void* buf, int length)\n",
    "{\n",
    "  int written = 0;\n",
    "  while (written < length)\n",
    "  {\n",
    "    int nw = ::write(fd, static_cast<const char*>(buf) + written, length - written);\n",
    "    if (nw > 0)\n",
    "    {\n",
    "      written += nw;\n",
    "    }\n",
    "    else if (nw == 0)\n",
    "    {\n",
    "      break;  // EOF\n",
    "    }\n",
    "    else if (errno != EINTR)\n",
    "    {\n",
    "      perror(\"write\");\n",
    "      break;\n",
    "    }\n",
    "  }\n",
    "  return written;\n",
    "}\n",
    "\n",
    "// 阻塞IO: 阻塞在read或者write, 彼此独立\n",
    "void run(TcpStreamPtr stream)\n",
    "{\n",
    "  // Caution: a bad example for closing connection\n",
    "  // 网络线程\n",
    "  std::thread thr([&stream] () {\n",
    "    char buf[8192]; // 8k缓冲区\n",
    "    int nr = 0;\n",
    "    // 从socket读取 \n",
    "    while ( (nr = stream->receiveSome(buf, sizeof(buf))) > 0) \n",
    "    {\n",
    "      int nw = write_n(STDOUT_FILENO, buf, nr); // 发送到stdout\n",
    "      if (nw < nr)\n",
    "      {\n",
    "        break;\n",
    "      }\n",
    "    }\n",
    "\n",
    "    // 读socket返回0, stdout不能shutdown, shutdown也无用, stdout和stdin是独立的, 不能阻止主线程继续从stdin读数据, \n",
    "    // 即对方关闭了socket连接, 主线程还在stdin中读取数据. 所以采用exit(0)退出进程. \n",
    "    // 主线程阻塞在stdin, 没有好的办法通知或者唤醒主线程, 如果采用IO复用, 就不会有这个问题, 因为只有一个线程, 就不需要唤醒. go语言中一般是关闭channel\n",
    "    // 或者将主线程改为select, 再加一个用于线程之间通知的wake up, event fd;\n",
    "\n",
    "    ::exit(0);  // should somehow notify main thread instead\n",
    "  });\n",
    "\n",
    "  // 主线程 \n",
    "  char buf[8192];\n",
    "  int nr = 0;\n",
    "  // 主线程反复读8k\n",
    "  while ( (nr = ::read(STDIN_FILENO, buf, sizeof(buf))) > 0)\n",
    "  {\n",
    "    int nw = stream->sendAll(buf, nr); // 发送到socket\n",
    "    if (nw < nr) // 读stdin, 返回0, while循环退出\n",
    "    {\n",
    "      break;\n",
    "    }\n",
    "  }\n",
    "  stream->shutdownWrite(); // 关闭socket的write; 即对方read返回0, 等待对方关闭socket, 网络线程中read返回0, 网络线程跳出循环, 终止进程\n",
    "  thr.join();\n",
    "}\n",
    "\n",
    "int main(int argc, const char* argv[])\n",
    "{\n",
    "  if (argc < 3)\n",
    "  {\n",
    "    printf(\"Usage:\\n  %s hostname port\\n  %s -l port\\n\", argv[0], argv[0]);\n",
    "    return 0;\n",
    "  }\n",
    "\n",
    "  int port = atoi(argv[2]);\n",
    "  if (strcmp(argv[1], \"-l\") == 0) // 参数设置当做服务端\n",
    "  {\n",
    "    std::unique_ptr<Acceptor> acceptor(new Acceptor(InetAddress(port))); // accept一个连接\n",
    "    TcpStreamPtr stream(acceptor->accept());\n",
    "    if (stream)\n",
    "    {\n",
    "      acceptor.reset();  // stop listening\n",
    "      run(std::move(stream)); // 调用run\n",
    "    }\n",
    "    else\n",
    "    {\n",
    "      perror(\"accept\");\n",
    "    }\n",
    "  }\n",
    "  else  // 参数设置当做客户端\n",
    "  {\n",
    "    InetAddress addr;\n",
    "    const char* hostname = argv[1];\n",
    "    if (InetAddress::resolve(hostname, port, &addr))\n",
    "    {\n",
    "      TcpStreamPtr stream(TcpStream::connect(addr));\n",
    "      if (stream)\n",
    "      {\n",
    "        run(std::move(stream));\n",
    "      }\n",
    "      else\n",
    "      {\n",
    "        printf(\"Unable to connect %s\\n\", addr.toIpPort().c_str());\n",
    "        perror(\"\");\n",
    "      }\n",
    "    }\n",
    "    else\n",
    "    {\n",
    "      printf(\"Unable to resolve %s\\n\", hostname);\n",
    "    }\n",
    "  }\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 如何通知另外的线程退出? 另外的线程可能阻塞在read(或者write, accept).\n",
    "- 进程退出: 读stdin, 返回0; 读socket, 返回0"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 阻塞IO是自动限速的: 如果stdout的速度慢, 则读socket也会慢, 起到一个节流限速,之后将socket代理服务器, 还会讲到这个"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# IO-multiplexing"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- IO-multiplexing is also called event-driven, event-based, reactor\n",
    "  - It is not asynchronous!"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- thread-per-connection: 适用于连接数目少, 或者线程非常廉价的情况, 每个连接占用一个线程\n",
    "- 用户态线程API大概1993年\n",
    "- IO复用(一个进程处理多个文件描述符, 复用的事线程, thread of control), 分为两个阵营BSD的select和ATNA的poll, TCP/IP是BSD4.2引入, select也是BSD4.2引入(引入可能为了解决并发网络编程); IO-multiplexing: 事件驱动, 基于事件, 面向对象下的reactor, 是同步的, 不是异步的"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> C++20后可以使用协程"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- Use only one thread\n",
    "  - Stock netcats are implemented in this way"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 复用的线程, 一个线程可以处理多个socket fd"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- IO复用, 通常与no-blocking IO一起使用"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**例子: IO-multiplexing 为什么与 no-blocking 一起使用**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "python"
    }
   },
   "outputs": [],
   "source": [
    "#!/usr/bin/python\n",
    "\n",
    "# https://github.com/chenshuo/recipes/blob/master/python/netcat.py\n",
    "\n",
    "import os\n",
    "import select\n",
    "import socket\n",
    "import sys\n",
    "\n",
    "def relay(sock):\n",
    "    poll = select.poll()\n",
    "    # 注册2个读事件\n",
    "    poll.register(sock, select.POLLIN)\n",
    "    poll.register(sys.stdin, select.POLLIN)\n",
    "\n",
    "    # python通过GC自动关闭, 不需要手动socket close\n",
    "\n",
    "    done = False\n",
    "    # IO复用通常是一个巨大的事件循环\n",
    "    while not done:\n",
    "        events = poll.poll(10000)  # 10 seconds\n",
    "        for fileno, event in events: # 等待事件, fileno文件描述符\n",
    "            if event & select.POLLIN: # 读事件\n",
    "                if fileno == sock.fileno():\n",
    "                    data = sock.recv(8192)\n",
    "                    if data:\n",
    "                        sys.stdout.write(data) # 写入stdout\n",
    "                    else:\n",
    "                        done = True # 读到0, 程序退出\n",
    "                else:\n",
    "                    assert fileno == sys.stdin.fileno()\n",
    "                    data = os.read(fileno, 8192)\n",
    "                    if data:\n",
    "                        sock.sendall(data)\n",
    "                    else: # 读到0字节\n",
    "                        # 需要等待数据发送完\n",
    "                        sock.shutdown(socket.SHUT_WR)\n",
    "                        poll.unregister(sys.stdin)\n",
    "\n",
    "\n",
    "def main(argv):\n",
    "    if len(argv) < 3:\n",
    "        binary = argv[0]\n",
    "        print \"Usage:\\n  %s -l port\\n  %s host port\" % (argv[0], argv[0])\n",
    "        return\n",
    "    port = int(argv[2])\n",
    "    if argv[1] == \"-l\": # 服务端\n",
    "        # server\n",
    "        server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)\n",
    "        server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)\n",
    "        server_socket.bind(('', port))\n",
    "        server_socket.listen(5)\n",
    "        (client_socket, client_address) = server_socket.accept()\n",
    "        server_socket.close()\n",
    "        relay(client_socket)\n",
    "    else: # 客户端\n",
    "        # client\n",
    "        sock = socket.create_connection((argv[1], port))\n",
    "        relay(sock)\n",
    "\n",
    "\n",
    "if __name__ == \"__main__\":\n",
    "    main(sys.argv)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- With blocking IO: one direction could block the other"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "程序存在的问题: 两个方向, 发送方和接受方的数据传输不是相互独立的, 用的阻塞IO, 就有真的阻塞的可能"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> 1993年 Solaris 2.2, Windows NT 3.1开始支持多线程程序; \n",
    "\n",
    "> 在Socket之前, 一个UNIX进程也可以处理多个文件描述符, 一个进程可以打开多个磁盘文件, 读写这些文件, 做一个外部排序;\n",
    "\n",
    "> 为什么BSD 4.2在引入TCP协议栈和Socket API的同时, 引入select系统调用, IO复用这个编程模型呢? 主要原因: 磁盘文件对应的文件描述符是个被动. 一个程序可以自行决定什么时候读写, 不会遇到阻塞的情况. 阻塞的意思是, 你要去读, 但是没有数据可以读. 文件在那, 要么数据可读, 要么读到文件末尾(end of file), 不会等待.\n",
    "\n",
    "> 在磁盘上的阻塞, 还有另外一个含义, 磁盘操作一直都是阻塞的, 会等待读完, 才会返回, 在aio之前, 没有办法异步读.\n",
    "\n",
    "> 一个程序可以打开多个磁盘文件, 程序自己决定读/写哪个文件; 而Socket文件描述符是主动的, 它是一个流(stream). Socket什么时候有数据可读, 取决于对方发数据, 这是跟磁盘文件的根本性的不同.\n",
    "\n",
    "> 如果有一个进程同时打开了多个Socket文件描述符, 它不能自己决定读取哪个, 否则读A阻塞,那么B到来, 就读取不了了. 所以BSD为了解决这个引入IO复用, 在读之前查一下, 现在有哪些socket可读. 如果不引入IO复用, 很难想象程序处理多个Socket描述符的并发情况. 使用fork-per-connection旧的并发模型无法处理2个以上的并发连接. 使用fork-per-connection洗衣歌http server, 每一个连接之间是独立的, 但是写一个聊天服务就困难了\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- Demo\n",
    "  - Server: chargen.py, a white hole\n",
    "  - Client: nc, nc < /dev/zero"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- a white hole: 只发数据, 不收数据\n",
    "- nc: 使用不同版本的nc, 区别在于nc有没有从标准输入读, 如果chargen只发数据, 不收数据的话.如果nc从标准输入, 有数据读, 发到Socket上后, 数据会很快填满缓冲区, 填满数据后, 就发不了数据了, 发不了数据之后会有什么现象nc如何实现的."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 实验*"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 系统的`nc`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "```shell\n",
    "./chargen -l 1234\n",
    "```\n",
    "\n",
    "```shell\n",
    "nc localhost 1234 > /dev/null\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 加一些输入\n",
    "\n",
    "```shell\n",
    "./chargen -l 1234\n",
    "```\n",
    "\n",
    "```shell\n",
    "nc localhost 1234  < /dev/zero > /dev/null\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 阻塞在write\n",
    "\n",
    "```shell\n",
    "./chargen -l 1234\n",
    "```\n",
    "\n",
    "```shell\n",
    "strace nc localhost 1234  < /dev/zero > /dev/null\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "```shell\n",
    "netstat -tpn | grep 1234\n",
    "```\n",
    "\n",
    "![4.3](./images/4.3.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- chargen有输入, 没有读\n",
    "- nc输出缓冲区满了, 没有写"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### [recipes/python/netcat.py](https://github.com/chenshuo/recipes/blob/master/python/netcat.py)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "```shell\n",
    "./chargen -l 1234\n",
    "```\n",
    "\n",
    "```shell\n",
    "python netcat.py localhost 1234 > /dev/null\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 加一些输入\n",
    "\n",
    "```shell\n",
    "./chargen -l 1234\n",
    "```\n",
    "\n",
    "```shell\n",
    "python netcat.py localhost 1234 < /dev/zero > /dev/null\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### [recipes/tpc/netcat.cc](https://github.com/chenshuo/recipes/blob/master/tpc/bin/netcat.cc) "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "```shell\n",
    "./chargen -l 1234\n",
    "```\n",
    "\n",
    "```shell\n",
    "netcat localhost 1234 > /dev/null\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 加一些输入: 吞吐量下降, 但是没有阻塞\n",
    "\n",
    "```shell\n",
    "./chargen -l 1234\n",
    "```\n",
    "\n",
    "```shell\n",
    "netcat localhost 1234 < /dev/zero >> /dev/null\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Non-blocking IO to rescue"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- Code is much more complicated, short read and short write\n",
    "\t- Caution: in this example we only show how todeal with short write"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 当IO操作阻塞时，会打断其他方向的输入输, 如何编写应用程序\n",
    "- 在已经有非阻塞IO实现库的情形下, 如何实现网络程序\n",
    "\n",
    "> 应用程序不应该与非阻塞IO打交道, 这是网络库的责任"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "python"
    }
   },
   "outputs": [],
   "source": [
    "#!/usr/bin/python\n",
    "\n",
    "# https://github.com/chenshuo/recipes/blob/master/python/netcat-nonblock.py\n",
    "\n",
    "import errno\n",
    "import fcntl\n",
    "import os\n",
    "import select\n",
    "import socket\n",
    "import sys\n",
    "\n",
    "def setNonBlocking(fd): # 设置非阻塞O_NONBLOCK\n",
    "    flags = fcntl.fcntl(fd, fcntl.F_GETFL)\n",
    "    fcntl.fcntl(fd, fcntl.F_SETFL, flags | os.O_NONBLOCK)\n",
    "\n",
    "\n",
    "def nonBlockingWrite(fd, data):\n",
    "    try:\n",
    "        nw = os.write(fd, data)\n",
    "        return nw\n",
    "    except OSError as e: # 将异常变回错误码\n",
    "        if e.errno == errno.EWOULDBLOCK:\n",
    "            return -1\n",
    "\n",
    "# 程序只展示了一半非阻塞写的内容, stdout使用阻塞, socket非阻塞\n",
    "def relay(sock): \n",
    "    socketEvents = select.POLLIN # socket两个事件不能单独注册, 所以要用一个变量保存\n",
    "    poll = select.poll()\n",
    "    poll.register(sock, socketEvents)\n",
    "    poll.register(sys.stdin, select.POLLIN) # 只关心输入事件\n",
    "\n",
    "    setNonBlocking(sock)\n",
    "    # setNonBlocking(sys.stdin)\n",
    "    # setNonBlocking(sys.stdout)\n",
    "\n",
    "    done = False\n",
    "    socketOutputBuffer = ''\n",
    "    while not done:\n",
    "        events = poll.poll(10000)  # 10 seconds\n",
    "        for fileno, event in events:\n",
    "            if event & select.POLLIN: # 输入事件\n",
    "                if fileno == sock.fileno(): # 当socket可读\n",
    "                    data = sock.recv(8192)\n",
    "                    if data:\n",
    "                        nw = sys.stdout.write(data)  # stdout does support non-blocking write, though\n",
    "                    else:\n",
    "                        done = True\n",
    "                else:\n",
    "                    assert fileno == sys.stdin.fileno() # stdin可读\n",
    "                    data = os.read(fileno, 8192)\n",
    "                    if data:\n",
    "                        assert len(socketOutputBuffer) == 0 # 要保证socketOutputBuffer中没有数据, 否则会出现乱序\n",
    "                        nw = nonBlockingWrite(sock.fileno(), data) # nw表示写了多少个字节\n",
    "                        if nw < len(data): # 没有写完, 发生了short write\n",
    "                            if nw < 0:\n",
    "                                nw = 0\n",
    "                            socketOutputBuffer = data[nw:] # 将socketOutputBuffer设置为剩余的数据\n",
    "                            socketEvents |= select.POLLOUT # 开始关注POLLOUT事件, 为什么不一直关注, 因为会造成busy loop, 电平触发的缺点\n",
    "                            poll.register(sock, socketEvents) # 重新注册socket\n",
    "                            poll.unregister(sys.stdin) # 不是一个通用的写法, 是一个应用特定(netcat), 暂时不读stdin, 其他应用程序可能不需要这样\n",
    "                    else: # 没有数据读取, 说明stdin读取完毕\n",
    "                        sock.shutdown(socket.SHUT_WR)\n",
    "                        poll.unregister(sys.stdin)\n",
    "            if event & select.POLLOUT: # 输出事件, 表示可写, 但是不能表示能写完, 非阻塞有可能发生short write\n",
    "                if fileno == sock.fileno():\n",
    "                    assert len(socketOutputBuffer) > 0 # socketOutputBuffer应该一个以上\n",
    "                    nw = nonBlockingWrite(sock.fileno(), socketOutputBuffer)\n",
    "                    # 仍然需要判断是否写完\n",
    "                    if nw < len(socketOutputBuffer): # 没有写完\n",
    "                        assert nw > 0\n",
    "                        socketOutputBuffer = socketOutputBuffer[nw:] # 应用层缓冲区交给内核缓冲区, 记录没有写完的数据\n",
    "                    else: # 如果写完了 nw 等于 len(socketOutputBuffer)\n",
    "                        socketOutputBuffer = ''\n",
    "                        socketEvents &= ~select.POLLOUT  # 停止观察POLLOUT事件, 否则会busy loop\n",
    "                        poll.register(sock, socketEvents)\n",
    "                        poll.register(sys.stdin, select.POLLIN) # netcat的特殊处理, 重新观察stdin\n",
    "\n",
    "\n",
    "\n",
    "def main(argv):\n",
    "    if len(argv) < 3:\n",
    "        binary = argv[0]\n",
    "        print \"Usage:\\n  %s -l port\\n  %s host port\" % (argv[0], argv[0])\n",
    "        print (sys.stdout.write)\n",
    "        return\n",
    "    port = int(argv[2])\n",
    "    if argv[1] == \"-l\":\n",
    "        # server\n",
    "        server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)\n",
    "        server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)\n",
    "        server_socket.bind(('', port))\n",
    "        server_socket.listen(5)\n",
    "        (client_socket, client_address) = server_socket.accept()\n",
    "        server_socket.close()\n",
    "        relay(client_socket)\n",
    "    else:\n",
    "        # client\n",
    "        sock = socket.create_connection((argv[1], port))\n",
    "        relay(sock)\n",
    "\n",
    "\n",
    "if __name__ == \"__main__\":\n",
    "    main(sys.argv)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "```shell\n",
    "./chargen -l 1234\n",
    "```\n",
    "\n",
    "```shell\n",
    "python netcat-nonblock.py localhost 1234 > /dev/null\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 非阻塞版本, 虽然有输入, socket不可写, 但是会继续从socket读数据\n",
    "\n",
    "```shell\n",
    "./chargen -l 1234\n",
    "```\n",
    "\n",
    "```shell\n",
    "python netcat-nonblock.py localhost 1234 < /dev/zero > /dev/null\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- Application level buffers are needed\n",
    "- Usually non-blocking IO is delegated to a good network library, like libevent, netty, muduo, etc."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 使用事件循环写程序逻辑很零散, 因为要求不能阻塞, 类似Windows事件循环, 阻塞后, 界面失去响应. 建议如果能用thread-per-connection, 尽量用, 这个模型贴近人的思维, 而IO复用,基于事件贴近机器的思维 性能好,但是写起来困难\n",
    "- 非阻塞IO通常是网络库管理, 明白其中的道理, 不需要自己去写, 配合网络库去写程序, 在回调中, 它告诉你事件数据到了, 你不要去做一些耗时的操作(阻塞在IO上), 例如, 事件可读发送响应的函数里去查数据库\n",
    "- 非阻塞IO的网络程序维护困难, 如果在接收后要进行很多步计算, 程序逻辑散在不同的回调函数中, 特别是在某一步中添加功能, 这个功能有可能阻塞的话, 你需要将函数拆为2个, 需要维护状态, 将阻塞的状态交给网络库处理, 保存线程恢复现场的功能.\n",
    "\n",
    "> 如何避免回调. go语言做的很好"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- ACE is a bad example, its Reactor only dispatches events, you have to write your own code for non-blocking read/write inevery program."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 没有讲非阻塞IO; 基于事件可以看ACE, 非阻塞IO就不要看ACE了"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Why non-blocking IO is a must"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 为什么要在IO复用编程模式下一定要使用非阻塞IO\n",
    "- 前面举的例子, 阻塞在write上, 是因为没有去检查pollout事件, 如果检查了, 事件可读再去读, 可写再去写, 这样使用阻塞IO也可以吗? 不可以"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Example from Unix Network Programming\n",
    "- Calling [`accept(2)`](https://man7.org/linux/man-pages/man2/accept.2.html) after a listening socket is \"ready for reading\" could block, because client could have disconnected in between."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- `accept(2)`: 如果有连接上来, 它会使文件描述符/server的socket变得可读, 但是在变得可读和准备accept之间, 客户端可能断开连接了, accept就可能会永远阻塞下去. 所以想使用IO复用模型处理这个accept, 那这个accept的server的socket应该是非阻塞的"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- [select(2)](https://man7.org/linux/man-pages/man2/select.2.html): Under Linux,`select()` may report a socket file descriptor as \"ready for reading\", while nevertheless a subsequent read blocks. This could for example happen when data has arrived but uponexamination has wrong checksum and is discarded. There may be other circumstances in which a file descriptor is spuriously reported as ready. Thus it may be safer to useO_NONBLOCK on sockets that should not block."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- select(2): select会将一个socket文件描述符说成是可读, 之后的一个read可能会阻塞, 就算告诉你可读了, 但是你真的去读时也有可能会阻塞"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# When short write happens in non-blocking IO"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 非阻塞的读, 在第二部分会仔细讲, 用到一个应用层的缓冲区, 读到的数据不全要先存到缓冲区里, 等凑够一条完整的消息, 再触发消息处理逻辑\n",
    "- 非阻塞的写, 在第二部分不会仔细的讲, 因为这是网络库的内容, 通常做的是告诉网络库, 你要发送多少数据 ,至于网络库如何处理事件, 如何触发POLLOUT, 什么时候可以发送, 这个是网络库处理的内容"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- Save remaining data in some buffer\n",
    "  - Never call write() when buffer is not empty,it reorders data\n",
    "  - Alternatively, always send from buffer"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 如果发数据, 发的不完整, 剩下的数据要放在某个缓冲区里, 而且缓冲区不是空的, 就不能对新的数据, 调用write(), 否则就会造成数据的乱序\n",
    "- 举例: 要发送1k的数据, 发送了500个byte, 剩下的500个byte放缓冲区里, 现在又要发送600个byte数据, 不能对着600个字节的数据调用write(), 要将这600个byte数据放在500byte待发数据的后面, 变成1.1k的数据, 从缓冲区里面发送\n",
    "- 始终从缓冲区里面发送, 每次发送之前都要看下POLLOUT事件; 现在的网络库一般不会这么做, 如果发送缓冲区是空, 就直接write, 写不完了, 才会丢到缓冲区里; 以前比较保守的程序, 发送数据, 先往应用层的发送缓冲区放, 然后注册POLLOUT事件, 然后处理POLLOUT事件, 从应用层的发送缓冲区, 往外发送数据"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- Start watching POLLOUT event\n",
    "  - Meanwhile, any write() should append the buffer instead"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 一旦发生short write, 即发送的数据没有发完, 需要向放松缓冲区填数据, 要观察POLLOUT事件, 同时在观察POLLOUT事件, 发送缓冲区不为空的期间, 别人调用你的网络库来发送数据时, 应该往发送缓冲区去append数据, 而不是write"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- When POLLOUT is ready, write from buffer\n",
    "  - Consume buffer"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 如果socket变得可写, POLLOUT事件在select/poll/epoll返回时, 就从发送缓冲区拿数据, 然后写到socket上, 同时缓冲区也会消费掉里面数据"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- If buffer becomes empty, stop watching POLLOUT event\n",
    "  - Otherwise, it end up with a busy loop"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 如果buffer变空了, 那就停止观察POLLOUT事件, 否则没有数据可发, 但是socket又是可写的, 用的电平触发, 就会busy loop, 100% CPU"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# What if sink is slow?"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- The common pitfall in non-blocking IO\n",
    "  - Avoid memory exhaustion or dropping messages\n",
    "- Stop reading andstop watching readiness\n",
    "  - Otherwise end up with a busy loop in level-triggered IO multiplexer\n",
    "- Sender will be throttled because TCP advertised window drops to zero\n",
    "- Amortized throughput will be the same\n",
    "  - Due to buffering, instant throughput will be different"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 非阻塞IO遇到的两一个问题: 对方接受数据缓慢"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "![4.4](./images/4.4.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 下水管道流出慢:\n",
    "  - 水位上升: 溢出丢失消息; 消耗过多的内存\n",
    "\n",
    "\n",
    "- 带宽不匹配的问题: \n",
    "  - 网关与内部是千兆网连接, 客户是10M带宽的家用宽带, 客户通过网关代理, 向内部请求一个大的数据, 由于客户带宽小, 只能慢慢发送\n",
    "  - 设置一个high work mark, 高过一个水位后停止发送, 等待下水管道流出低于low work mark, 再打开水龙头\n",
    "  - 有可能高于high work mark, 假设hwm为1M, 一次写了2M, lwm一般为0\n",
    "  - 当使用非阻塞IO, 写一个代理服务器, 在第七层工作, http proxy, 是可行的. 因为发送方, 会停下来发, 水龙头另外的一头, 停止读以后, TCP会有一个advice window, 告诉我这边可以读的数据, 等本机TCP协议栈接受数据满了以后, 给对方的source ack, window是0, 然后对方就把这个速度降下来了, 程序处理得当话\n",
    "  - echo: 如果使用非阻塞IO去实现, 有read事件就去读, 然后放到本机的缓冲区里面, 等write事件, 再去写. 如果对方一直就是光发不收怎么办? 一种是你拼命地去读, 然后不断地往缓冲区里面加, 内存被消耗完; 第二种, 如果写是阻塞的话, 就影响了echo上的其他客户端. 可行的办法是, 对echo来讲, 对于这个连接的发送缓冲区, 如果超过了一定的高水位(1M), 就停止去读. 简单的协议可以, 对于比较复杂的协议, 没有这么简单, 例如rpc, 发送过是一个很小的请求(get image), 发过去的响应很大, 如果不断地发请求, 服务端的数据很快就堆积起来, 发送缓冲区就会越积越大, 这个停止读, 好像是网络库干的事情不对了, 因为这个应该在rpc那层解决的\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Level-trigger and edge trigger?"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 非阻塞IO, IO复用: 使用电平触发(Level-trigger), 还是边缘触发(edge trigger)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- `select(2)` and `poll(2)` are level-trigger, community has 30+ years of experienceon how to write correct code, many 3rd party libraries rely on this"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 传统来讲, `select`, `poll`都是电平触发, 就是有数据, 就一直告诉你有"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- `epoll(5)` stands for edge-poll, works in both LT and ET mode"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- epoll的`e`指边缘触发, 但同时支持LT和ET模式, 最早是只支持ET, 后来发现可能替代select, poll的行为不一致,可能会导致一些问题"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- No up-to-date benchmarks show which is faster"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 目前没有新的性能对比, 比较哪个更快;"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- Edge trigger works best for writing and accepting\n",
    "- Level trigger works best for readingdata"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- ET: 从编写程序, 对发送数据, 接受新连接是最好的;\n",
    "- LT: 从读数据来说是最好的, 不会造成饥饿; 饥饿是数据没读完, 它不会再告诉你数据是可读的, 会等到下次一次对方发送数据到达时, 才会告诉你; 举例子, 对方发来1K的数据, 如果使用边缘触发, 它会告诉你有数据可读, 你要去读, 如果没有读完1K, 读了800bytes, 剩下200bytes还在缓冲区. 如果是电平触发, 你去poll的时候, 它会告诉你还有200bytes数据可读, 你马上会读取这200bytes数据. 如果是边缘触发, 它不会告诉你数据可读, 认为你已经读过了, 没有新的数据到来, 这200bytes数据直到对方发送新的数据到达时, 重新拿到数据可读事件. ET就是有事件变化的时候, 新的数据到达了, 才会触发一次. 触发一次后,它不管你读没读, 在第二次数据到达之前, 就不再触发了"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- Current Linux kernel doesn't support mix LT/ET for on socket\n",
    "  - Muduo uses level trigger"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 是否可以, 混用两种模式. ET应对write和accept, LT应对read. \n",
    "  - 如果发生文件描述符用尽的情况, accepting会陷入死循环, 这时候, 用ET是好的\n",
    "  - 目前Linux kernel没有区分读和写, 使用的同一个数据结构表示就绪事件\n",
    "  - Muduo: 使用LT, 为了和第三方库配合, 如果使用ET有可能造成丢事件, 使用LT告诉对方可读, 对方一定会读, 就不会造成死循环"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# IO-multiplexing in a multithreaded program"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- Topic of Part II of this course"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "C++17",
   "language": "C++17",
   "name": "xcpp17"
  },
  "language_info": {
   "codemirror_mode": "text/x-c++src",
   "file_extension": ".cpp",
   "mimetype": "text/x-c++src",
   "name": "C++17",
   "version": "17"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}
